package com.atguigu.gmall.publisher.service.impl;

import com.atguigu.gmall.publisher.bean.NameValue;
import com.atguigu.gmall.publisher.bean.OrderDetailResult;
import com.atguigu.gmall.publisher.service.OrderService;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.index.query.MatchQueryBuilder;
import org.elasticsearch.index.query.Operator;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.ParsedSum;
import org.elasticsearch.search.aggregations.metrics.SumAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


@Service
@Slf4j
public class OrderServiceImpl implements OrderService {


    @Autowired
    RestHighLevelClient restHighLevelClient;

    private final String indexNamePrefix = "gmall0701_order_wide_";

    @Override
    public List<NameValue> getStatsByItem(String itemName, String date, String type) {
        String indexName = indexNamePrefix + date;
        //1 构建条件
        //1.1 查询条件
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(new MatchQueryBuilder("sku_name", itemName).operator(Operator.AND));
        //1.2 聚合操作
        String field = typeToField(type);
        TermsAggregationBuilder groupAggs = AggregationBuilders.terms("groupby_" + field).field(field).size(100);
        SumAggregationBuilder sumAggs = AggregationBuilders.sum("sum_amount").field("split_total_amount");
        groupAggs.subAggregation(sumAggs);
        searchSourceBuilder.aggregation(groupAggs);
        //1.3 因为是聚合操作所以把明细部分清除
        searchSourceBuilder.size(0);


        SearchRequest searchRequest = new SearchRequest(indexName).source(searchSourceBuilder);
        try {
            //2 执行 查询
            SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            //3 提取聚合结果
            Aggregations aggregations = searchResponse.getAggregations();
            ParsedTerms groupbyTerms = aggregations.get("groupby_" + field);
            List<? extends Terms.Bucket> buckets = groupbyTerms.getBuckets();
            List<NameValue> nameValueList = new ArrayList<>();
            for (Terms.Bucket bucket : buckets) {
                String key = bucket.getKeyAsString();
                ParsedSum sumAmount = bucket.getAggregations().get("sum_amount");
                double value = sumAmount.getValue();
                NameValue nameValue = new NameValue(key, value);
                nameValueList.add(nameValue);
            }
            //返回前做转换
            return changNameValueByType(type, nameValueList);
        } catch (ElasticsearchStatusException ese) {
            if (ese.status() == RestStatus.NOT_FOUND) {
                log.warn("索引 未建立：" + ese.getIndex());
                return new ArrayList<>();
            }
            throw new RuntimeException("es 查询异常");
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("es 连接异常");
        }

    }

    @Override
    public OrderDetailResult getDetailByItem(String itemName, String date, Integer pageNo, Integer pageSize) {
        //构造查询条件
        String indexName = indexNamePrefix + date;
        // 执行查询

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        //1 匹配
        searchSourceBuilder.query(new MatchQueryBuilder("sku_name", itemName).operator(Operator.AND));
        //2 分页
        searchSourceBuilder.size(pageSize);
        searchSourceBuilder.from((pageNo-1)*pageSize);  //行号= （页码-1）*每页行数
        //3 抓取的字段
        searchSourceBuilder.fetchSource(new String[]{"create_time", "sku_name", "order_price", "province_name", "sku_num", "total_amount", "user_age", "user_gender"},null);
        //4 高亮显示
        searchSourceBuilder.highlighter(new HighlightBuilder().field("sku_name"));

        SearchRequest searchRequest = new SearchRequest(indexName).source(searchSourceBuilder);
        try {
            SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            // 提取从查询结果
            //提取总值
            long total = searchResponse.getHits().getTotalHits().value;
            //提取明细
            SearchHit[] hits = searchResponse.getHits().getHits();
            List<Map> sourceList=new ArrayList<>();
            for (SearchHit hit : hits) {
                Map<String, Object> sourceAsMap = hit.getSourceAsMap();
                //提取高亮字段
                Map<String, HighlightField> highlightFieldMap = hit.getHighlightFields();
                HighlightField highlightField = highlightFieldMap.get("sku_name");
                Text text = highlightField.getFragments()[0];
                String skuNameWithHL = text.string();
                //把高亮字段覆盖原字段
                sourceAsMap.put("sku_name",skuNameWithHL);

                sourceList.add(sourceAsMap);
            }
            //封装到结果对象中
            OrderDetailResult orderDetailResult = new OrderDetailResult(total, sourceList);
            return orderDetailResult;

        } catch (ElasticsearchStatusException ese) {
            if (ese.status() == RestStatus.NOT_FOUND) {
                log.warn("索引 未建立：" + ese.getIndex());
                return new OrderDetailResult();
            }
            throw new RuntimeException("es 查询异常");
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException("es 连接异常");
        }



    }

    //根据业务类型 调整结果
    private List<NameValue> changNameValueByType(String type, List<NameValue> nameValueList) {
        if (type.equals("gender")) {
            for (NameValue nameValue : nameValueList) {
                if (nameValue.getName().equals("M")) {
                    nameValue.setName("男");
                } else {
                    nameValue.setName("女");
                }
            }
            return nameValueList;
        } else if (type.equals("age")) {  //作业：根据业务调整结果
            return nameValueList;
        }
        return nameValueList;
    }

   //根据业务类型 选择聚合字段
    private String typeToField(String type) {
        if (type.equals("gender")) {
            return "user_gender";
        } else if (type.equals("age")) {
            return "user_age";
        } else {
            return null;

        }

    }
}
